package com.letv.dataprocess;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.letv.bean.SummaryBean;
import com.letv.bean.VideoBean;
import com.letv.common.ConnectionManager;
import com.letv.common.Constants;
import com.letv.common.ProcessData;
import com.letv.conf.ConfigFactory;
import com.letv.download.DownloadUtils;
import com.letv.mms.cache.factory.CacheFactory;
import com.letv.relatedstar.StarLoading;
import com.letv.util.DaquanAop;
import com.letv.util.DateUtil;
import com.letv.util.SQLUtil;
import com.letv.util.SoKeyUtil;
import com.letv.util.StringUtil;
import com.letv.util.DaquanAop.DaquanOfflineMark;

/**
 * 汇聚接口
 * 1：注拼音
 * 2：下载过程中没有的字段 补全，如  播放量
 *
 */
public class DataprocessServiceImpl implements DataprocessService {

    /**
     * DownloadUtils, common method for download data from cms
     */
    private DownloadUtils downloadUtils = new DownloadUtils();

	private final static Logger log = LoggerFactory.getLogger(DataprocessServiceImpl.class);

	private static String cbasekeyprefix = ConfigFactory.getString("cbaseprefix.name");


	@Override
    public SummaryBean process(String type) {
		log.info("begin process " + type + "...");
		SummaryBean bean = new SummaryBean();

		//get DBconnection
		ConnectionManager cm = null;
		try {
			cm = ConnectionManager.getInstance();
		} catch (Exception e) {
			log.error("connection get instance error!", e);
			bean.setErrorMsg("connection get instance error! " + e.getMessage());
			return bean;
		}
        Connection conn = cm.getConnection();

        log.info("create index_" + type + " table...");
        if (Constants.INDEX_LESO.equals(type)) {
            this.downloadUtils.copyTable(conn, Constants.LESO_INDEX_TB, Constants.LESO_OLD_INDEX_TB);
        } else if (Constants.INDEX_ALBUM.equals(type)) {
            this.downloadUtils.copyTable(conn, Constants.ALBUM_INDEX_TB, Constants.ALBUM_OLD_INDEX_TB);
        } else if (Constants.INDEX_STAR.equals(type)) {
            this.downloadUtils.copyTable(conn, Constants.STAR_INDEX_TB, Constants.STAR_OLD_INDEX_TB);
        } else if (Constants.INDEX_VIDEO.equals(type)) {
            this.downloadUtils.copyTable(conn, Constants.VIDEO_INDEX_TB, Constants.VIDEO_OLD_INDEX_TB);
        } else {
            log.warn(type + " data is not exist!");
        }
		//get minId and maxId
		ArrayList<String> idRange = new ArrayList<String>();

		try {
			if (Constants.INDEX_LESO.equals(type)) {
				idRange = SQLUtil.getIdRange(conn, Constants.LESO_TB, "");
	        } else if (Constants.INDEX_ALBUM.equals(type)) {
	        	idRange = SQLUtil.getIdRange(conn, Constants.ALBUM_TB, "");
	        } else if (Constants.INDEX_STAR.equals(type)) {
	        	idRange = SQLUtil.getIdRange(conn, Constants.STAR_TB, "");
	        } else if (Constants.INDEX_VIDEO.equals(type)) {
	        	idRange = SQLUtil.getIdRange(conn, Constants.VIDEO_TB, "");
	        } else if (Constants.INDEX_LESO_EXTRAET.equals(type)) {
	        	Connection connLeso = cm.getConnectionFromLeso();
	        	try{
		        	idRange = SQLUtil.getIdRange(connLeso, Constants.LESO_EXTRAET_SRC_TB, "");
	        	}catch(SQLException e){
	        		log.error("get " + type + " id range error!", e);
	    			bean.setErrorMsg("get " + type + " id range error! " + e.getMessage());
	    			return bean;
	        	}finally{
	                if(!SQLUtil.close(connLeso)) {
	                    bean.setErrorMsg(bean.getErrorMsg() + " [ close db connection error! ");
	                }
	        	}
	        }
		} catch (Exception e) {
			log.error("get " + type + " id range error!", e);
			bean.setErrorMsg("get " + type + " id range error! " + e.getMessage());
			return bean;
		} finally {
		    if(!SQLUtil.close(conn)) {
		        bean.setErrorMsg(bean.getErrorMsg() + " [ close db connection error! ");
		    }
		}
		if (Constants.INDEX_STAR.equals(type)) {
			if (!StarLoading.load()) {
				log.error("load relate star file failed");
			}
		}

		int minId = Integer.parseInt(idRange.get(0)) - 1;
		int maxId = Integer.parseInt(idRange.get(1));

		int poolSize = ConfigFactory.getInt("threadPool.basesize", 30);
		log.info("DataProcessWorker thread pool size:{}", poolSize);
        ExecutorService threadPool = Executors.newFixedThreadPool(poolSize);
		List<DataProcessWorker> workList = new ArrayList<DataProcessWorker>();
		List<Future<DataProcessResult>> results = new ArrayList<Future<DataProcessResult>>();

		int pageSize = ConfigFactory.getInt("thread.pageSize", 3000);
		int totalPage = 0;
		totalPage = (maxId - minId) / pageSize + 1;
		log.info("type="+type+",totalPage=" + totalPage);

		for (int i = 1; i <= totalPage; i++) {
			DataProcessWorker worker = null;

			if (Constants.INDEX_LESO.equals(type)) {
				worker = new DataProcessWorker(Constants.LESO_TB, Constants.LESO_INDEX_TB, type, minId + pageSize * (i - 1), minId + pageSize * i);
	        } else if (Constants.INDEX_ALBUM.equals(type)) {
	        	worker = new DataProcessWorker(Constants.ALBUM_TB, Constants.ALBUM_INDEX_TB, type, minId + pageSize * (i - 1), minId + pageSize * i);
	        } else if (Constants.INDEX_STAR.equals(type)) {
	        	worker = new DataProcessWorker(Constants.STAR_TB, Constants.STAR_INDEX_TB, type, minId + pageSize * (i - 1), minId + pageSize * i);
	        } else if (Constants.INDEX_VIDEO.equals(type)) {
	        	worker = new DataProcessWorker(Constants.VIDEO_TB, Constants.VIDEO_INDEX_TB, type, minId + pageSize * (i - 1), minId + pageSize * i);
	        } else if (Constants.INDEX_LESO_EXTRAET.equals(type)) {
	        	worker = new DataProcessWorker(Constants.LESO_EXTRAET_SRC_TB, Constants.LESO_EXTRAET_SRC_TB, type, minId + pageSize * (i - 1), minId + pageSize * i);
	        }

			workList.add(worker);
		}

		//get result of threads
		int successCnt = 0;
		int failCnt = 0;
		int cbaseSuccessCnt = 0;
		int cbaseFailCnt = 0;

		try {
			results = threadPool.invokeAll(workList);

			for (Future<DataProcessResult> future : results) {
				try {
					DataProcessResult r = future.get();
					if (!future.isCancelled()) {
						successCnt += r.getRecordsCnt();
						cbaseSuccessCnt += r.getCbaseSuccess();
						log.info("resolve " + r.getRecordsCnt() + " 条 数据, result=" + r.isFlag());
					} else {
						failCnt += r.getRecordsCnt();
						cbaseFailCnt += r.getCbasefail();
						log.info("cancelled! resolve " + r.getRecordsCnt() + " 条 数据, result=" + r.isFlag());
					}
				} catch (ExecutionException e) {
					log.error("get thread result ExecutionException!", e);
					bean.setErrorMsg(bean.getErrorMsg() + " [get thread result ExecutionException!" + e.getMessage());
					continue;
				} catch (Exception e2) {
					log.error("get thread result Exception!", e2);
					bean.setErrorMsg(bean.getErrorMsg() + " [ get thread result exception!" + e2.getMessage());
					continue;
				}
			}
		} catch (InterruptedException e1) {
			log.error("excute " + type + " thread error!", e1);
			bean.setErrorMsg(bean.getErrorMsg() + " [excute " + type + " thread error!" + e1.getMessage());
		} finally {
			threadPool.shutdown();
			log.info("process " + type + " threadPool shutdown!");
		}
		log.info("process " + type + " done!");
		bean.setSubTaskMysqlSuccessedCountNum(successCnt);
		bean.setSubTaskMysqlFailedCountNum(failCnt);
		bean.setSubTaskCbaseSuccessedCountNum(cbaseSuccessCnt);
		bean.setSubTaskCbaseFailedCountNum(cbaseFailCnt);
		return bean;
	}

    @Override
    public boolean processDataById(String type, Long id) {
		Connection conn = null;
		Connection connToMMS = null;
		try {
			ConnectionManager cm = ConnectionManager.getInstance();
	        conn = cm.getConnection();
	        connToMMS = cm.getConnectionFromMMS();

			HashMap<String, Object> map = new HashMap<String, Object>();
			ArrayList<Map<String, Object>> albumDetailMapList = new ArrayList<Map<String, Object>>();
			HashMap<String, Object> f = new HashMap<String, Object>();

			if (Constants.INDEX_ALBUM.equals(type)) {
				map = SQLUtil.getDataById(conn, Constants.ALBUM_TB, id);
			} else if (Constants.INDEX_STAR.equals(type)) {
				map = SQLUtil.getDataById(conn, Constants.STAR_TB, id);
			} else if (Constants.INDEX_VIDEO.equals(type)) {
				map = SQLUtil.getDataById(conn, Constants.VIDEO_TB, id);
			} else if (Constants.INDEX_LESO.equals(type)) {
				albumDetailMapList = SQLUtil.getProcessData(conn, Constants.LESO_TB, id.intValue(), id.intValue() + 1);
			}

			try {
				f = SQLUtil.getIndexFieldName(conn, "index_" + type);
			} catch (Exception e) {
				log.error("get " + type + " index fieldname error!", e);
			}

			if (map == null) {
				log.info("data type=" + type + ", id=" + id + " is not satisfied!");
				return false;
			}

			List<String> recordList = new ArrayList<String>();
			if (Constants.INDEX_ALBUM.equals(type)) {
				Map<String, Object> m = ProcessData.processAlbum(map, connToMMS);
				recordList.add(ProcessData.getIndexAlbumStr(m));

				if (f.size() != 0) {
					Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "album");
					putCbase(Constants.SRC_VRS, indexMap, "inc","album");
				} else {
					log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
				}
			} else if (Constants.INDEX_STAR.equals(type)) {
				Map<String, Object> m = ProcessData.processStar(map);
				recordList.add(ProcessData.getIndexStarStr(m));

				if (f.size() != 0) {
					Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "star");
					putCbase(Constants.SRC_VRS, indexMap, "inc","star");
				} else {
					log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
				}
			} else if (Constants.INDEX_VIDEO.equals(type)) {
				Map<String, Object> m = ProcessData.processVideo(map);
				recordList.add(ProcessData.getIndexVideoStr(m));

				if (f.size() != 0) {
					Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "video");
					putCbase(Constants.SRC_VRS, indexMap, "inc","video");
				} else {
					log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
				}
			} else if (Constants.INDEX_LESO.equals(type)) {
			    for(int i = 0; i < albumDetailMapList.size(); i++) {
	                Map<String, Object> m = ProcessData.processLeso(albumDetailMapList.get(i));
	                Object vid= m.get("vid");
	                if ( vid != null && StringUtils.isNotBlank(vid.toString())) {
	                    recordList.add(ProcessData.getIndexLesoStr(m));
	                }
	                if (f.size() != 0) {
	                    Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "leso");
	                    putCbase(Constants.SRC_LESO, indexMap, "inc", "album");
	                    this.putLetvMappingToCbase(m);
	                } else {
	                    log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
	                }			        
			    }
			}
			if (null != recordList && recordList.size() > 0) {
			    for(String record : recordList) {
			        if(!StringUtils.isBlank(record)) {
			            SQLUtil.insertSingleIndexData(conn, type, record);
			        }
			    }
			}
			return true;
		} catch (Exception e) {
			log.error("insert index_" + type + " data id=" + id + " error!", e);
			return false;
		} finally {
		    SQLUtil.close(conn);
		    SQLUtil.close(connToMMS);
		}
	}

    public static void processClip(Map<String, Object> map) {
        if (Constants.WEBSITE_LETV.equals(map.get("source"))
                || String.valueOf(Constants.LESO_ZONGYI).equals(map.get("category"))) {
            Connection connLeso = null;
            try {
                ConnectionManager cm = ConnectionManager.getInstance();
                connLeso = cm.getConnectionFromLeso();
                // ArrayList<Map<String, Object>> videolists =
                // SQLUtil.getLesoAlbumVideoSourceData(connLeso, "leso.SQL3",
                // vid,"180001");
            } catch (Exception e) {

            }
        }
    }

    @Override
    public boolean delDataById(String type, Long id) {
		ConnectionManager cm = null;
        Connection conn = null;
		try {
			cm = ConnectionManager.getInstance();
			conn = cm.getConnection();

			if(Constants.INDEX_LESO.equals(type)) {
			    this.deleteLesoById(conn, id);
			} else {
	            SQLUtil.deleteDataById(conn, "index_" + type, id);
	            deleteFromCbase(type, id);			    
			}
			log.info("delete type=index_" + type + ", id=" + id + " success!");
			return true;
		} catch (Exception e) {
			log.error("delete index_" + type + " data error! id=" + id, e);
			return false;
		} finally {
            SQLUtil.close(conn);
		}
	}

    /**
     * 从index_leso表及cbase中删除外网专辑
     * @param connectionFromLocal Connection, 与离线计算本地数据库的连接
     * @param albumId, 外网专辑id
     */
	private void deleteLesoById(Connection connectionFromLocal, Long albumId) {
        if(null == connectionFromLocal || null == albumId) {
            return;
        }
        try {
            List<String> otherDetailSourceList = SQLUtil.getOtherDetailSourceList(connectionFromLocal, albumId.toString());
            log.info(JSON.toJSONString("other source:" + otherDetailSourceList));
            SQLUtil.deleteDataById(connectionFromLocal, "index_" + Constants.INDEX_LESO, albumId);
            if(null == otherDetailSourceList || otherDetailSourceList.size() == 0) {
                this.deleteFromCbase("index_" + Constants.INDEX_LESO, albumId);
            } else {
                this.deleteLesoFromCbase(albumId, otherDetailSourceList);
            }
        } catch(Exception e) {
            log.error("error to delete leso album by id,id=" + albumId, e);
        }
    }

	/**
	 * 当包含非默认站点源时从index_leso中删除专辑
	 * @param albumId Long, 外网专辑id
	 * @param otherDetailSourceList List<String>, 非默认站点源source列表
	 */
    private void deleteLesoFromCbase(Long albumId, List<String> otherDetailSourceList) {
        List<String> keyList = new ArrayList<String>();
        String key = SoKeyUtil.getSOKey(Constants.SRC_LESO, Constants.LESO, albumId.toString());
        keyList.add(key);
        for(String source : otherDetailSourceList) {
            keyList.add(key + "_" + source); 
        }
        for(String k : keyList) {
            k = cbasekeyprefix + k;
            CacheFactory.getCache().delete(k);
            log.info("delete key from cbase Ok, the key is " + k);
            k = "inc" + k;
            CacheFactory.getCache().delete(k);
            log.info("delete key from cbase Ok, the key is " + k);
        }
    }

    private void deleteFromCbase(String type, Long id) {
	    String key = null;
	    String sid = id.toString();
	    if (type.equals(Constants.INDEX_LESO)) {
	        key = SoKeyUtil.getSOKey(Constants.SRC_LESO, Constants.LESO, sid);
	    } else if (type.equals(Constants.INDEX_ALBUM)) {
	        key = SoKeyUtil.getSOKey(Constants.SRC_VRS, Constants.ALBUM, sid);
	    } else if (type.equals(Constants.INDEX_VIDEO)) {
	        key = SoKeyUtil.getSOKey(Constants.SRC_VRS, Constants.VIDEO, sid);
	    } else if (type.equals(Constants.INDEX_STAR)) {
	        key = SoKeyUtil.getSOKey(Constants.SRC_VRS, Constants.STAR, sid);
	    }
	    if ( key != null ) {
	        key = cbasekeyprefix+key;
	        CacheFactory.getCache().delete(key);
	        log.info("delete key from cbase Ok, the key is " + key);
	        key = "inc" + key;
	        CacheFactory.getCache().delete(key);
	        log.info("delete key from cbase Ok, the key is " + key);
	    }
    }

    @Override
    public boolean updDataById(String type, Long id) {
        if (delDataById(type, id)) {
        	return processDataById(type, id);
        } else {
			log.error("delete index_" + type + " data  id=" + id +" return false.");
			return false;
        }
	}

	private String getKey(String src, Map<String, Object> m) {
		String key = SoKeyUtil.getSoKey(src, m);
		return key;
	}

	protected boolean putCbase(String src, List<Map<String, Object>> data) {
		boolean flag = true;
		for (Map<String, Object> record : data) {
			String key = getKey(src, record);
			try {
				String jsonStr = JSONObject.toJSONString(record);
				CacheFactory.getCache().set(key, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));

				if (log.isDebugEnabled()) {
					log.debug("write key to cbase, the key is " + key);
				}
			} catch(Exception e) {
				log.error("write key to cbase error! the key is "+ key);
				flag = false;
			}
		}
		return flag;
	}

	protected boolean putCbase(String src, Map<String, Object> record, String flag, String type) {
	    String key = "";
		try {
		    key = getKey(src, record);
		    if("0".equals(String.valueOf(record.get("if_default_source")))) {
		        key = key + "_" + String.valueOf(record.get("subSrc"));
		    }
		    if ("inc".equals(flag)) {
		        key = "inc"+cbasekeyprefix+key;
		    }else {
		        key = cbasekeyprefix+key;
		    }
			String jsonStr = JSONObject.toJSONString(record);
			String id = "";
			if ("album".equals(type)) {
				id = String.valueOf(record.get("aid"));
			} else if ("video".equals(type)) {
				id = String.valueOf(record.get("vid"));
			} else if ("star".equals(type)) {
				id = String.valueOf(record.get("sid"));
			}
			log.info("id:"+id+",key : " + key+ ", jsonStr.size : " + jsonStr.length());
			if ("inc".equals(flag)) {
			    if(!Constants.IF_DAQUAN) {
			        CacheFactory.getCache().set(key, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 25));
			    }
			} else {
			    if(Constants.IF_DAQUAN) {
			        List<String> daquanKeyList = new ArrayList<String>();
			        daquanKeyList.add(key);
			        DaquanAop.addToDaquanLog(daquanKeyList, jsonStr, DaquanOfflineMark.TOTAL_DETAIL);
			    } else {
			        CacheFactory.getCache().set(key, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
			        CacheFactory.getCache().delete("inc"+key);
			    }
			}
 			log.info("id:"+id+",write key to cbase, the key is: " + key);
			return true;
		} catch (Exception e) {
			log.error("write key to cbase error! the key is: " + key);
			return false;
		}
	}

	/**
	 * 将媒资pid与外网id的对应关系存入cbase。
	 * key   : 媒资全局id--110_xxx (若该专辑无letv源则不会存入cbase)
	 * value : 为map结构：
	 *         leso_album_id:205_xxx
	 * @param record Map<String, Object>, leso数据详情
	 */
	protected void putLetvMappingToCbase(Map<String, Object> record) {
	    if(null == record) {
	        return;
	    }
        String source = StringUtil.convertObjectToString(record.get("source"));
        String leso_album_id = StringUtil.convertObjectToString(record.get("id"));
        String letv_album_id = StringUtil.convertObjectToString(record.get("letv_aid"));
        if(StringUtils.isBlank(source) || StringUtils.isBlank(leso_album_id) || StringUtils.isBlank(letv_album_id) || !Constants.WEBSITE_LETV.equals(source)) {
            return;
        }
	    try {
	        String key = new StringBuffer().append("mapping").append(Constants.VRS_ALBUM_PERFIX).append(letv_album_id).toString();
	        Map<String, String> value = new HashMap<String, String>();
	        value.put("leso_album_id", leso_album_id);
	        String jsonString = JSONObject.toJSONString(value);
            if(Constants.IF_DAQUAN) {
                List<String> daquanKeyList = new ArrayList<String>();
                daquanKeyList.add(key);
                DaquanAop.addToDaquanLog(daquanKeyList, jsonString, DaquanOfflineMark.TOTAL_DETAIL);
            } else {
                CacheFactory.getCache().set(key, jsonString, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
            }
	        log.info("success to put letv mapping to cbase!leso_album_id=" + leso_album_id + ",key=" + key);
	    } catch(Exception e) {
	        log.error("put letv mapping to cbase error!leso_album_id=" + leso_album_id + ",letv_album_id=" + letv_album_id, e);
	    }
	}

	public boolean processLesoExtraetById(int id, String action) {
		try {
			if (Constants.MQ_ACTION_ADD.equals(action) || Constants.MQ_ACTION_MOD.equals(action)) {
				ConnectionManager cm = null;
				try {
					cm = ConnectionManager.getInstance();
				} catch (Exception e) {
					return false;
				}
		        Connection conn = cm.getConnectionFromLeso();
				Map<String, Object> record = new HashMap<String, Object>();
				try {
					ArrayList<Map<String, Object>> lists = SQLUtil.getProcessNewLesoWebsiteData(conn, id, id+1);
					//record = SQLUtil.getSourceDataById(conn, Constants.INDEX_LESO_EXTRAET, id);
					if (null == lists || lists.isEmpty() || 0 == lists.size()) {
						log.error("lesoExtraet data is not exist! id=" + id);
						return false;
					}
					record = lists.get(0);
					String oldkey = (cbasekeyprefix.indexOf("2")>0?"2":"")+"leso_ea_" + record.get("leso_aid") + "_" + record.get("source");
					String key = cbasekeyprefix+"leso_ea_" + record.get("leso_aid") + "_" + record.get("source");
					ArrayList<Map<String, Object>> videolists = null;
					String category = record.get("video_type").toString();
					HashMap<String, Object> record1 = new HashMap<String, Object>();
					String vid = record.get("id").toString();
					videolists = SQLUtil.getLesoAlbumVideoSourceData(conn, "leso.SQL3", vid,"180001");
					if( videolists ==null || videolists.size()  == 0) {
						log.info("single_video don't have data,video_id="+vid);
						return false;
					}
				    String jsonStr = DataProcessWorker.getWebSiteJSON(record, record1, videolists, category, key);
					CacheFactory.getCache().set(oldkey, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
					CacheFactory.getCache().set(key, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
					if ("letv".equals(record.get("source")) && record.get("vrs_aid") != null) {
						String key11 ="leso_la_" + record.get("vrs_aid");
						log.info(key+",inc website write key to cbase, leso_la_ key is " + (cbasekeyprefix.indexOf("2")>0?"2":"")+key11);
						CacheFactory.getCache().set((cbasekeyprefix.indexOf("2")>0?"2":"")+key11, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
						CacheFactory.getCache().set(cbasekeyprefix + key11, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
					}
					log.info("write key to cbase, the key is " + key);
					return true;
				} catch (Exception e) {
					log.error("get lesoExtraet data error! id=" + id, e);
					return false;
				} finally {
		            SQLUtil.close(conn);
				}

			} else {
				CacheFactory.getCache().delete((cbasekeyprefix.indexOf("2")>0?"2":"")+action);
				log.info("website delete key from cbase, the key is " + (cbasekeyprefix.indexOf("2")>0?"2":"")+action);
				CacheFactory.getCache().delete(cbasekeyprefix+action);
				log.info("website delete key from cbase, the key is " +cbasekeyprefix+ action);
				return true;
			}
		} catch(Exception e) {
			if (Constants.MQ_ACTION_ADD.equals(action) || Constants.MQ_ACTION_MOD.equals(action)) {
				log.warn("write key to cbase error! the key is "+ id);
			} else {
				log.warn("delete key from cbase error! the key is "+ id);
			}
			return false;
		}
	}

	public boolean processLesoExtraetById_leso(int id, String action) {
		try {
			if (Constants.MQ_ACTION_ADD.equals(action) || Constants.MQ_ACTION_MOD.equals(action)) {
				ConnectionManager cm = null;
				try {
					cm = ConnectionManager.getInstance();
				} catch (Exception e) {
					return false;
				}
		        Connection conn = cm.getConnectionFromLeso();

				Map<String, Object> record = new HashMap<String, Object>();

				try {
					ArrayList<Map<String, Object>> lists = SQLUtil.getProcessNewLesoWebsiteData(conn, id, id+1);
					//record = SQLUtil.getSourceDataById(conn, Constants.INDEX_LESO_EXTRAET, id);
					if (null == lists || lists.isEmpty() || 0 == lists.size()) {
						log.error("lesoExtraet data is not exist! id=" + id);
						return false;
					}
					record = lists.get(0);
					String oldkey = (cbasekeyprefix.indexOf("2")>0?"2":"")+"leso_ea_" + record.get("leso_aid") + "_" + record.get("source");
					String key = cbasekeyprefix+"leso_ea_" + record.get("leso_aid") + "_" + record.get("source");
					ArrayList<Map<String, Object>> videolists = null;
					String category = record.get("video_type").toString();
					HashMap<String, Object> record1 = new HashMap<String, Object>();
					String vid = record.get("id").toString();
					videolists = SQLUtil.getLesoAlbumVideoSourceData(conn, "leso.SQL3", vid,"180001");
					if( videolists ==null || videolists.size()  == 0) {
						log.info("single_video don't have data,video_id="+vid);
						return false;
					}
					String jsonStr = DataProcessWorker.getWebSiteJSON(record, record1, videolists, category, key);
					CacheFactory.getCache().set(oldkey, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
					CacheFactory.getCache().set(key, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
					if ("letv".equals(record.get("source")) && record.get("vrs_aid") != null) {
						String key11 ="leso_la_" + record.get("vrs_aid");
						log.info(key+",inc website write key to cbase, leso_la_ key is " + (cbasekeyprefix.indexOf("2")>0?"2":"")+key11);
						CacheFactory.getCache().set((cbasekeyprefix.indexOf("2")>0?"2":"")+key11, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
						CacheFactory.getCache().set(cbasekeyprefix + key11, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
					}
					log.info("write key to cbase, the key is " + key);
					return true;
				} catch (Exception e) {
					log.error("get lesoExtraet data error! id=" + id, e);
					return false;
				} finally {
		            SQLUtil.close(conn);
				}

			} else {
				CacheFactory.getCache().delete((cbasekeyprefix.indexOf("2")>0?"2":"")+action);
				log.info("website delete key from cbase, the key is " + (cbasekeyprefix.indexOf("2")>0?"2":"")+action);
				CacheFactory.getCache().delete(cbasekeyprefix+action);
				log.info("website delete key from cbase, the key is " +cbasekeyprefix+ action);
				return true;
			}
		} catch(Exception e) {
			if (Constants.MQ_ACTION_ADD.equals(action) || Constants.MQ_ACTION_MOD.equals(action)) {
				log.warn("write key to cbase error! the key is "+ id);
			} else {
				log.warn("delete key from cbase error! the key is "+ id);
			}
			return false;
		}
	}

	@Override
    public SummaryBean processWebsite(String type) {
		log.info("begin process " + type + "...");
		SummaryBean bean = new SummaryBean();

		//get DBconnection
		ConnectionManager cm = null;
		try {
			cm = ConnectionManager.getInstance();
		} catch (Exception e) {
			log.error("connection get instance error!", e);
			bean.setErrorMsg("connection get instance error! " + e.getMessage());
			return bean;
		}
        Connection conn = cm.getConnection();

		//get minId and maxId
		ArrayList<String> idRange = new ArrayList<String>();

		try {
			if (Constants.INDEX_LESO_EXTRAET.equals(type)) {
	        	Connection connLeso = cm.getConnectionFromLeso();
	        	try{
		        	idRange = SQLUtil.getIdRangeNewLesoWebsite(connLeso);
	        	}catch(SQLException e){
	        		log.error("get " + type + " id range error!", e);
	    			bean.setErrorMsg("get " + type + " id range error! " + e.getMessage());
	    			return bean;
	        	}finally{
	                if(!SQLUtil.close(connLeso)) {
	                    bean.setErrorMsg(bean.getErrorMsg() + " [ close db connection error! ");
	                }
	        	}
	        }
		} catch (Exception e) {
			log.error("get " + type + " id range error!", e);
			bean.setErrorMsg("get " + type + " id range error! " + e.getMessage());
			return bean;
		} finally {
            if(!SQLUtil.close(conn)) {
                bean.setErrorMsg(bean.getErrorMsg() + " [ close db connection error! ");
            }
		}

		int minId = Integer.parseInt(idRange.get(0)) - 1;
		int maxId = Integer.parseInt(idRange.get(1));

		ExecutorService threadPool = Executors.newFixedThreadPool(ConfigFactory.getInt("threadPool.basesize", 30));
		List<DataProcessWorker> workList = new ArrayList<DataProcessWorker>();
		List<Future<DataProcessResult>> results = new ArrayList<Future<DataProcessResult>>();

		int pageSize = ConfigFactory.getInt("thread.pageSize", 3000);
		int totalPage = 0;
		totalPage = (maxId - minId) / pageSize + 1;
		log.info("totalPage=" + totalPage);

		for (int i = 1; i <= totalPage; i++) {
			DataProcessWorker worker = null;

			if (Constants.INDEX_LESO_EXTRAET.equals(type)) {
	        	worker = new DataProcessWorker(Constants.LESO_EXTRAET_SRC_TB, Constants.LESO_EXTRAET_SRC_TB, type, minId + pageSize * (i - 1), minId + pageSize * i);
	        }

			workList.add(worker);
		}
//		Integer[] tempVideoIdList = {1, 2148985};
//		for(Integer tempVideoId : tempVideoIdList) {
//		    DataProcessWorker worker = new DataProcessWorker(Constants.LESO_EXTRAET_SRC_TB, Constants.LESO_EXTRAET_SRC_TB, type, tempVideoId, tempVideoId + 1);
//		    workList.add(worker);
//		}

		//get result of threads
		int successCnt = 0;
		int failCnt = 0;
		int cbaseSuccessCnt = 0;
		int cbaseFailCnt = 0;

		try {
			results = threadPool.invokeAll(workList);

			for (Future<DataProcessResult> future : results) {
				try {
					DataProcessResult r = future.get();
					if (!future.isCancelled()) {
						successCnt += r.getRecordsCnt();
						cbaseSuccessCnt += r.getCbaseSuccess();
						log.info("resolve " + r.getRecordsCnt() + " 条 数据, result=" + r.isFlag());
					} else {
						failCnt += r.getRecordsCnt();
						cbaseFailCnt += r.getCbasefail();
						log.info("cancelled! resolve " + r.getRecordsCnt() + " 条 数据, result=" + r.isFlag());
					}
				} catch (ExecutionException e) {
					log.error("get thread result ExecutionException!", e);
					bean.setErrorMsg(bean.getErrorMsg() + " [get thread result ExecutionException!" + e.getMessage());
					continue;
				} catch (Exception e2) {
					log.error("get thread result Exception!", e2);
					bean.setErrorMsg(bean.getErrorMsg() + " [ get thread result exception!" + e2.getMessage());
					continue;
				}
			}
		} catch (InterruptedException e1) {
			log.error("excute " + type + " thread error!", e1);
			bean.setErrorMsg(bean.getErrorMsg() + " [excute " + type + " thread error!" + e1.getMessage());
		} finally {
			threadPool.shutdown();
			log.info("process " + type + " threadPool shutdown!");
		}

		log.info("process " + type + " done!");
		bean.setSubTaskMysqlSuccessedCountNum(successCnt);
		bean.setSubTaskMysqlFailedCountNum(failCnt);
		bean.setSubTaskCbaseSuccessedCountNum(cbaseSuccessCnt);
		bean.setSubTaskCbaseFailedCountNum(cbaseFailCnt);
		return bean;
    }

	@Override
    public SummaryBean processSmallWebsite(String type) {
		log.info("begin process " + type + "...");
		SummaryBean bean = new SummaryBean();
		//get DBconnection
		ConnectionManager cm = null;
		try {
			cm = ConnectionManager.getInstance();
		} catch (Exception e) {
			log.error("connection get instance error!", e);
			bean.setErrorMsg("connection get instance error! " + e.getMessage());
			return bean;
		}

    	Connection connLeso = cm.getConnectionFromLeso();
    	Map<String, List<Map<String, Object>>> maps = new HashMap<String, List<Map<String,Object>>>();
    	try{
    		List<Map<String, Object>> lists = SQLUtil.getProcessSmallWebsiteData(connLeso, Constants.LESO_EXTRAET_SRC_TB);
    		for (Map<String, Object> record : lists) {

    			String key = cbasekeyprefix+"leso_ea_small" + record.get("leso_aid");
				HashMap<String, Object> record1 = new HashMap<String, Object>();

				record1.put("site", record.get("source"));
				record1.put("aid", record.get("leso_aid"));
				record1.put("vrsAid", record.get("vrs_aid"));
				record1.put("videoPlayUrls", record.get("playurls"));
				record1.put("orderNum", record.get("top_hot"));
				record1.put("isPc", record.get("pc"));
				record1.put("isTv", record.get("tv"));
				record1.put("isMobile", record.get("mobile"));
				record1.put("category", record.get("video_type"));
				record1.put("playStreams", record.get("quality"));
				record1.put("tophot", record.get("top_hot"));
				record1.put("episodes", record.get("episodes"));
				record1.put("episodeStatus", record.get("episode_status"));
				record1.put("nowEpisodes", record.get("currentepisode"));
				record1.put("intro", record.get("intro"));
				record1.put("dataType", String.valueOf(Constants.ALBUM));

				Object playurl = record.get("playurls");
				if (null != playurl) {
					String videoPlayUrls = playurl.toString();
					List<VideoBean> videoList = new ArrayList<VideoBean>();
					String[] videoArray = videoPlayUrls.split(";");
					String[] info = record1.get("intro")!=null ? record1.get("intro").toString().split(";"): new String[0];
					for (int i = 0; i < videoArray.length; i++) {
						if (StringUtils.isNotBlank(videoArray[i])) {
							String[] videoInfo = info.length > i ? info[i].split("\\|") : new String[0];
							String[] videoFileds = videoArray[i].split("\\|");
							if (videoFileds.length == 2) {
								VideoBean video = new VideoBean();
								video.setDataType(Constants.VIDEO + "");
								video.setAorder(StringUtils.isNotBlank(videoFileds[0])?videoFileds[0].replace("-", ""):"");
								video.setUrl(videoFileds[1]);
								video.setName(videoInfo.length > 1 ? videoInfo[1] : "精彩内容点击观看。");
								videoList.add(video);
							}
						}
					}
					if (videoList.size() > 0) {
						record1.put("videoList", videoList);
					}
				}

				if (!maps.containsKey(key)) {
					List<Map<String, Object>> teList = new ArrayList<Map<String,Object>>();
					teList.add(record1);
                }else {
                	List<Map<String, Object>> teList = maps.get(key);
                	teList.add(record1);
                	maps.put(key, teList);
				}
            }
    	}catch(SQLException e){
    		log.error("get " + type + " id range error!", e);
			bean.setErrorMsg("get " + type + " id range error! " + e.getMessage());
			return bean;
    	}finally{
            if(!SQLUtil.close(connLeso)) {
                bean.setErrorMsg(bean.getErrorMsg() + " [ close db connection error! ");
            }
    	}

    	int successCnt = 0;
		int failCnt = 0;
		int cbaseSuccessCnt = 0;
		int cbaseFailCnt = 0;

		Set<String> keySet = maps.keySet();
		for (String key : keySet) {
			List<Map<String, Object>> teList = maps.get(key);
			String jsonStr = JSONObject.toJSONString(teList);
			CacheFactory.getCache().set(key, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
			successCnt++;
			cbaseSuccessCnt++;
        }

		log.info("process " + type + " done!");
		bean.setSubTaskMysqlSuccessedCountNum(successCnt);
		bean.setSubTaskMysqlFailedCountNum(failCnt);
		bean.setSubTaskCbaseSuccessedCountNum(cbaseSuccessCnt);
		bean.setSubTaskCbaseFailedCountNum(cbaseFailCnt);
		return bean;
    }
}